package com.yanggu.bigdata.realtime.app.dwd

import com.yanggu.bigdata.realtime.app.OutputTagObject.{displayOutputTag, errorJsonDataOutputTag, startLogOutputTag}
import com.yanggu.bigdata.realtime.app.function.{OdsBaseLogToDwdSideOutputProcessFunction, ParseJsonStrProcessFunction, SetNewUserFlagRichMapFunction}
import com.yanggu.bigdata.realtime.common.GmallConfig.KAFKA_BROKER_LIST
import com.yanggu.bigdata.realtime.utils.KafkaUtil
import org.apache.flink.streaming.api.scala._

//数据流: 前端埋点数据 -> nginx -> 日志服务器 -> Kafka(ods_base_log) -> Flink(ods->dwd) -> Kafka(dwd_log)
//程 序:  mock-log -> nginx -> logger.sh -> Kafka(zk) -> Flink(BaseLogApp) -> Kafka(dwd_start_log、dwd_page_log、dwd_display_log)
object BaseLogApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val jsonObjectDataStream = KafkaUtil.getKafkaDataStream(env, KAFKA_BROKER_LIST, "ods_base_log", "ods_dwd_base_log_app")
      //将json字符串解析成JsonObject, 同时进行catch, 把错误数据进行测流输出
      .process(new ParseJsonStrProcessFunction)

    //将解析失败的数据在控制台进行打印输出
    jsonObjectDataStream.getSideOutput(errorJsonDataOutputTag).printToErr("DirtyData>>>>>>>")

    val pageDataStream = jsonObjectDataStream
      //由于需要使用状态需要进行keyBy
      .keyBy(_.getJSONObject("common").getString("mid"))
      .map(new SetNewUserFlagRichMapFunction)
      //将数据按照日志类型进行分流。页面数据输出到主流, 启动日志输出到测输出流, 曝光日志输出到测输出流
      .process(new OdsBaseLogToDwdSideOutputProcessFunction)

    //获取启动日志流并打印
    //pageDataStream.getSideOutput(startLogOutputTag).print("Start>>>>>>")
    pageDataStream.getSideOutput(startLogOutputTag)
      .addSink(KafkaUtil.getKafkaSink(KAFKA_BROKER_LIST, "dwd_start_log"))
    //获取曝光日志流并打印
    //pageDataStream.getSideOutput(displayOutputTag).print("Display>>>>>>")
    pageDataStream.getSideOutput(displayOutputTag)
      .addSink(KafkaUtil.getKafkaSink(KAFKA_BROKER_LIST, "dwd_display_log"))
    //页面日志流打印
    //pageDataStream.print("Page>>>>>")
    pageDataStream.addSink(KafkaUtil.getKafkaSink(KAFKA_BROKER_LIST, "dwd_page_log"))

    env.execute("BaseLogApp Job")
  }

}
